package com.shujia.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{Column, DataFrame, SparkSession}

object Demo5Burk {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("Demo5Burk")
      .config("spark.sql.shuffle.partitions", 2)
      .getOrCreate()

    // 导入隐式转换
    import spark.implicits._
    // 导入Spark SQL所有的函数
    import org.apache.spark.sql.functions._

    // 读取数据
    val burkDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("burk String,year String,tsl01 String,tsl02 String,tsl03 String,tsl04 String,tsl05 String,tsl06 String,tsl07 String,tsl08 String,tsl09 String,tsl10 String,tsl11 String,tsl12 String")
      .load("Spark/data/burks.txt")

    //    burkDF.show(10)

    burkDF.createOrReplaceTempView("burk")

    spark.sql(
      """
        |select  t1.burk
        |        ,t1.year
        |        ,t1.month
        |        ,t1.income
        |        ,sum(t1.income) over(partition by t1.burk,t1.year order by t1.month) as sum_income
        |from (
        |    select  burk
        |            ,year
        |            ,month
        |            ,income
        |    from burk lateral view explode(map(1,tsl01,2,tsl02,3,tsl03,4,tsl04,5,tsl05,6,tsl06,7,tsl07,8,tsl08,9,tsl09,10,tsl10,11,tsl11,12,tsl12)) v1 as month,income
        |)t1
        |""".stripMargin)
    //.show(30)


    val m: Column = map(
      expr("1"), $"tsl01",
      expr("2"), $"tsl02",
      expr("3"), $"tsl03",
      expr("4"), $"tsl04",
      expr("5"), $"tsl05",
      expr("6"), $"tsl06",
      expr("7"), $"tsl07",
      expr("8"), $"tsl08",
      expr("9"), $"tsl09",
      expr("10"), $"tsl10",
      expr("11"), $"tsl11",
      expr("12"), $"tsl12")

    /**
      * withColumn： 在前面DAtaFrame的基础上上增加新的列
      *
      */

    /**
      *
      * 1、统计每个公司每年按月累计收入  行转列 --> sum窗口函数
      *
      */

    burkDF
      .select($"burk", $"year", explode(m) as Array("month", "income"))
      .withColumn("inc_income", sum($"income") over Window.partitionBy($"burk", $"year").orderBy($"month"))
    //.show()

    /**
      * 2、计算每一年每个月占一年收入的比例
      *
      */

    burkDF
      .select($"burk", $"year", explode(m) as Array("month", "income"))
      .withColumn("sum_income", sum($"income") over Window.partitionBy($"burk", $"year"))
      .withColumn("p", round($"income" / $"sum_income", 4))
    // .show()


    /**
      * coalesce: 返回第一个不为null的列
      *
      */
    /**
      *
      * 2、统计每个公司当月比上年同期增长率  行转列 --> lag窗口函数
      * 公司代码,年度,月度,增长率（当月收入/上年当月收入 - 1）
      *
      */
    burkDF
      .select($"burk", $"year", explode(m) as Array("month", "income"))
      .withColumn("lest_income", lag($"income", 1, 0) over Window.partitionBy($"burk", $"month").orderBy($"year"))
      .withColumn("p", coalesce($"income" / $"lest_income" - 1, expr("1.0")))
      .show(100)
  }

}
